package com.tj.bigdata.syncblock.shedule;

import com.tj.bigdata.syncblock.constant.Constant;
import com.tj.bigdata.syncblock.constant.Web3jConstants;
import com.tj.bigdata.syncblock.strategy.StreamSourceAndEsImpl;
import com.tj.bigdata.syncblock.strategy.SyncBlockStrategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.math.BigInteger;
import java.util.Objects;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author Administrator
 */
@Slf4j
@Component
public class SyncSchedule {

    @Value("${spring.syncblock.begin}")
    private String beginPro;

    private SyncBlockStrategy syncBlockStrategy;

    private StringRedisTemplate stringRedisTemplate;

    private volatile BigInteger beginBlock = new BigInteger("0");

    LongAdder retry = new LongAdder();
    LongAdder non = new LongAdder();

    @Autowired
    public void setSyncBlockStrategy(@Qualifier("streamSourceAndEsImpl") SyncBlockStrategy syncBlockStrategy) {
        this.syncBlockStrategy = syncBlockStrategy;
    }

    @Autowired
    public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Async("threadPoolExecutor")
    @Scheduled(cron = "0 0 */1 * * ?")
    public void syncBlock() {
        if (beginBlock.compareTo(BigInteger.ZERO) == 0) {
            beginBlock = new BigInteger(beginPro);
        }
        BigInteger blockNew = new BigInteger("0");
        BigInteger blockNew2 = new BigInteger("0");
        try {
            //当前节点最新区块
            blockNew = StreamSourceAndEsImpl.web3.ethBlockNumber().send().getBlockNumber();
            Thread.sleep(10000);
            blockNew2 = StreamSourceAndEsImpl.web3.ethBlockNumber().send().getBlockNumber();
        } catch (InterruptedException | IOException e) {
            e.printStackTrace();
        }
        if (blockNew.equals(blockNew2)) {
            //触发条件 节点变化
            non.increment();
            log.info("没有重试 {} 最新节点 {}", non.longValue(), blockNew);
            return;
        }
        //获取到的最新标志位
        BigInteger begin = new BigInteger(Objects.requireNonNull(stringRedisTemplate.opsForValue().get(Web3jConstants.BLOCK_FLAG)));
        int res = blockNew.compareTo(begin);
        //最新和上次同步比较，
        if (res > 0) {
            //自旋等待
            BigInteger flagBlock;
            LongAdder times = new LongAdder();
            if (!beginBlock.equals(new BigInteger(beginPro))) {
                //不是第一次执行
                do {
                    flagBlock = new BigInteger(Objects.requireNonNull(stringRedisTemplate.opsForValue().get(Web3jConstants.BLOCK_FLAG)));
                    try {
                        log.info("====自旋等待上次同步结束 上次任务游标 {}===", flagBlock);
                        times.increment();
                        Thread.sleep(60000);
                        //达到一定次数此次任务作废
                        if (times.longValue() == Constant.WAITTIMES) {
                            if (begin.equals(flagBlock)) {
                                log.info("执行上次故障任务 begin {}", begin);
                                beginBlock = begin;
                                break;
                            }
                            log.info("退出任务");
                            return;
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log.info("flagBlock redis最新 {} beginBlock 上次任务起点 {}", flagBlock, beginBlock);
                    //退出条件 上次任务达到预期终点 即下次本次起点
                } while (!flagBlock.equals(beginBlock));
            }
            retry.increment();
            //触发条件 节点变化  节点最新大于当前
            log.info("触发重试 {}  最新节点 {} 运行中，当前 {}", retry.longValue(), blockNew, beginBlock);
            //上次的终点作为起点
            syncBlockStrategy.batchBlockchain(beginBlock, blockNew);
            //重置起点
            beginBlock = blockNew;
            log.info("beginBlock {} ===", blockNew);
        } else {
            non.increment();
            log.info("没有重试 {} 最新节点 {} 运行中，当前 {}，", non.longValue(), blockNew, beginBlock);
        }
    }
}
